package com.yichat.micro.msghttpservice.core.base;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

/**
 * Created by chenmohan on 2019/04/23
 *
 * @description
 */
@Slf4j
public class AdvancedExecutor extends ThreadPoolExecutor {

    private static final String requestExecutorName = "requestExecutor";

    private static final String recordExecutorName = "recordExecutor";
    // 线程池名称,一般以业务名称命名,方便区分
    private String poolName;

    @Getter
    private LongAdder activeThreadNumber = new LongAdder();

    public AdvancedExecutor(String poolName, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory factory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, factory, handler);
        this.poolName = poolName;
    }

    /**
     * 任务执行之前,记录任务开始时间
     */
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        activeThreadNumber.increment();
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        activeThreadNumber.decrement();

// 统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止
        /*
        LOGGER.info(String.format(this.poolName
                        + "-pool-monitor: Duration: %d ms, PoolSize: %d, CorePoolSize: %d, Active: %d, Completed: %d, Task: %d, Queue: %d, LargestPoolSize: %d, MaximumPoolSize: %d,KeepAliveTime: %d, isShutdown: %s, isTerminated: %s",
                diff, this.getPoolSize(), this.getCorePoolSize(), this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount(),
                this.getQueue().size(), this.getLargestPoolSize(), this.getMaximumPoolSize(), this.getKeepAliveTime(TimeUnit.MILLISECONDS),
                this.isShutdown(), this.isTerminated()));
                */
    }

    public static ThreadPoolExecutor newRequestExecutor() {
        return new AdvancedExecutor(
                requestExecutorName,
                //return new ThreadPoolExecutor(
                16
                , 32
                , 10
                , TimeUnit.MINUTES
                , new LinkedBlockingDeque<>(2048)
                , new EventThreadFactory(requestExecutorName)
                , new LogCallerPolicy(requestExecutorName)
        );
    }

    public static ThreadPoolExecutor newRecordExecutor() {
        return new AdvancedExecutor(
                recordExecutorName,
                2
                , 8
                , 10
                , TimeUnit.MINUTES
                , new LinkedBlockingDeque<>(1024)
                , new EventThreadFactory(recordExecutorName)
                , new LogCallerPolicy(recordExecutorName)
        );
    }

    @Data
    @AllArgsConstructor
    public static class LogCallerPolicy extends ThreadPoolExecutor.CallerRunsPolicy {

        private String name;

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            log.warn("[AbstractHttpCaller] task数量超过线程池排队队列,立即执行溢出的task");
            super.rejectedExecution(r, e);
        }
    }


    /**
     * 生成线程池所用的线程,只是改写了线程池默认的线程工厂,传入线程池名称,便于问题追踪
     */
    static class EventThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        /**
         * 初始化线程工厂
         *
         * @param poolName 线程池名称
         */
        EventThreadFactory(String poolName) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = poolName + "-pool-" + poolNumber.getAndIncrement() + "-thread-";
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

}
